package com.dongzili.binlog.common.binlogsynchronization;

import com.dongzili.binlog.common.config.redis.RedisConfig;
import com.dongzili.binlog.common.config.redis.RedisOperationsUtil;
import com.dongzili.binlog.common.config.redis.StreamMessageListenerConfig;
import com.github.shyiko.mysql.binlog.BinaryLogClient;
import com.github.shyiko.mysql.binlog.event.*;
import com.mysql.cj.jdbc.MysqlDataSource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.core.annotation.Order;
import org.springframework.data.redis.connection.stream.ObjectRecord;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.connection.stream.StreamRecords;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.GenericJackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.io.Serializable;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import static com.github.shyiko.mysql.binlog.event.EventType.*;

@Order(5)
@Slf4j
@Component
//@AllArgsConstructor
public class MysqlBinLogListener implements BinaryLogClient.EventListener {

    @Resource
    BinLogConstants binLogConstants;

    @Resource
    private  MysqlDataSource mysqlDataSource;

    @Resource
    RedisTemplate<String, BinLogItem> redisTemplate;

    private final Map<Long, TableColum> dbTableIdCols = new HashMap<>();
    private final Map<String, Map<String, Colum>> dbTableCols = new HashMap<>();

    @Override
    public void onEvent(Event event) {
        EventType eventType = event.getHeader().getEventType();
//        redisService.setCacheObjectAsync(BinLogServer.binLogName, new BinLogProgress(binaryLogClient.getBinlogFilename(), binaryLogClient.getBinlogPosition()), 6000L);
        if (eventType == QUERY) {
            //处理表结构修改
            QueryEventData data = event.getData();
            String sql = data.getSql();
            String head = "ALTER TABLE `";
            if (sql.startsWith(head)) {
                int start = sql.indexOf(head) + head.length();
                int end = sql.indexOf("` \r\n");
                if (start < end) {
                    String substring = sql.substring(start, end);
                    String tableName = substring.replaceAll("`", "");
                    if (tableName.contains(".")) {
                        List<String> table = binLogConstants.getTable();
                        int index = table.indexOf(tableName);
                        if (index >= 0) {
                            String[] split = table.get(index).split("\\.");
                            dbTableCols.put(tableName, BinLogUtils.getColMap(mysqlDataSource, split[0], split[1]));
                        }

                    }
                }
            }
        } else if (eventType == EventType.TABLE_MAP) {
            //映射表结构
            List<String> table = binLogConstants.getTable();
            TableMapEventData tableData = event.getData();
            String dbTable = BinLogUtils.getDBTable(tableData.getDatabase(), tableData.getTable());
            long tableId = tableData.getTableId();
            if (dbTableCols.containsKey(dbTable)) {
                dbTableIdCols.put(tableId, new TableColum(tableId, dbTable, dbTableCols.get(dbTable)));
                dbTableCols.remove(dbTable);
            } else if (table.contains(dbTable) && !dbTableIdCols.containsKey(tableId)) {
                String[] split = table.get(table.indexOf(dbTable)).split("\\.");
                Map<String, Colum> colMap = BinLogUtils.getColMap(mysqlDataSource, split[0], split[1]);
                dbTableIdCols.put(tableId, new TableColum(tableId, dbTable, colMap));
            }
        } else if (isWrite(eventType) || isUpdate(eventType) || isDelete(eventType)) {
            // 只处理添加删除更新三种操作
            if (isWrite(eventType)) {
                WriteRowsEventData data = event.getData();
                for (Serializable[] row : data.getRows()) {
                    setBaseTable(data.getTableId(), row, eventType);
                }
            } else if (isUpdate(eventType)) {
                UpdateRowsEventData data = event.getData();
                for (Map.Entry<Serializable[], Serializable[]> row : data.getRows()) {
                    setBaseTable(data.getTableId(), row, eventType);
                }
            } else if (isDelete(eventType)) {
                DeleteRowsEventData data = event.getData();
                for (Serializable[] row : data.getRows()) {
                    setBaseTable(data.getTableId(), row, eventType);
                }
            }
        }
    }


    @SuppressWarnings("unchecked")
    private void setBaseTable(long tableId, Object row, EventType eventType) {
        if (dbTableIdCols.containsKey(tableId)) {
            TableColum tableColum = dbTableIdCols.get(tableId);
            String dbTable = tableColum.getTableName();
            Map<String, Colum> columMap = tableColum.getColMap();
//                Map<String, Colum> columMap = new HashMap<>();
            BinLogItem item = isUpdate(eventType)
                    ? new BinLogItem((Map.Entry<Serializable[], Serializable[]>) row, columMap, eventType)
                    : new BinLogItem((Serializable[]) row, columMap, eventType);
            item.setDbTable(dbTable);
            item.setServerId(tableId);
            item.setTimestamp(System.currentTimeMillis());

            Map<String, BinLogItem> content = new HashMap<>();
            content.put("binLogItem", item);
            log.info("binlog消息队列-表名:{},类型:{}", dbTable, eventType);
            ObjectRecord<String, BinLogItem> objectRecord = StreamRecords.newRecord()
                    .ofObject(item)
                    .withStreamKey(StreamMessageListenerConfig.streamKey)
                    .withId(RecordId.autoGenerate());

            redisTemplate.setHashValueSerializer(new StringRedisSerializer());
            RecordId recordId = redisTemplate.opsForStream().add(objectRecord);
            log.info("消息发送recordId:{}", recordId);

        }

    }
}
